import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import java.io.IOException;
import java.io.InputStream;
import java.sql.*;
import java.util.*;

public class DataAsync {
    static HashMap<String,String> tableNameMap = new HashMap<>();
    static HashMap<String, String> configMap;
    static HashMap<String, String> specialTableMap;
    static Connection targetConnection = null;
    static Connection originConnection = null;
    static PreparedStatement statementTable;
    static PreparedStatement statement;
    static ResultSet resultSetTable;
    static ResultSet resultSetTableData;
    static List<String> haveSyncTableNameList = new ArrayList<>();
    static String currentDealTableName;//当前处理的表
    static long currentDealTableRowNumer;//需要同步的数据总条数
    static long startTime;//计时器-开始时间
    static long endTime;//计时器-结束时间
    static int commitPageSize;//每页提交数据条数
    static int currentCommitPageSize;//每页提交数据条数

    public static void main(String []args){
        Timer timer = new Timer();
        try {
            //加载配置
            loadProperties();
            //同步进度查询
            timer.schedule(new ChechScheduleTimerTask(),5*1000L,5*1000L);
            //开始进行同步
            startSync();
        }
        catch (Exception e){
            e.printStackTrace();
        }
        finally {
            if(!haveSyncTableNameList.isEmpty()){
                System.out.print("\r");
                System.out.println("本次已同步表："+ String.join(",",haveSyncTableNameList));
            }
            //关闭定时任务
            timer.cancel();
        }
    }

    //加载配置
    public static void loadProperties() throws ClassNotFoundException, SQLException, IOException {
        configMap = new HashMap<>();
        Properties prop = new Properties();
        InputStream in = Class.class.getResourceAsStream("/jdbc.properties");
        prop.load(in);     ///加载属性列表
        for (String key : prop.stringPropertyNames()) {
            configMap.put(key, prop.getProperty(key));
        }
        assert in != null;
        in.close();

        String tablename = configMap.get("notDealTableName");
        String[] split = tablename.split(",");
        for(String sting:split){
            tableNameMap.put(sting,"");
        }
        commitPageSize = Integer.parseInt(configMap.get("commitPageSize"));

        //加载特殊表配置
        specialTableMap = new HashMap<>();
        Properties specialTableProp = new Properties();
        specialTableProp.load(Class.class.getResourceAsStream("/special_table.properties"));     ///加载属性列表
        for (String key : specialTableProp.stringPropertyNames()) {
            specialTableMap.put(key, specialTableProp.getProperty(key));
        }

        Class.forName(configMap.get("classname"));
        originConnection = DriverManager.getConnection(configMap.get("originurl"), configMap.get("originuser"), configMap.get("originpassword"));// 用驱动类获取连接

        Class.forName("org.postgresql.Driver");
        targetConnection = DriverManager.getConnection(configMap.get("targeturl"), configMap.get("targetuser"), configMap.get("targetpassword"));// 用驱动类获取连接
        targetConnection.setAutoCommit(false);
    }
    
    //同步数据
    public static void startSync() throws Exception {
        String sql = configMap.get("sql");
        statementTable = originConnection.prepareStatement(sql);
        resultSetTable = statementTable.executeQuery();
        while(resultSetTable.next()){
            String table_name = resultSetTable.getString("table_name");
            syncTableData(table_name.toLowerCase());
        }
        System.out.print("\r");
        System.out.println("所有数据已同步完成");
        originConnection.close();
    }

    public static void syncTableData(String tableName) throws Exception {
        //排除不需要处理的表
        if (tableNameMap.containsKey(tableName)) return;
        boolean isDelete = "true".equals(configMap.get("isDelete"));
        if(!specialTableMap.containsKey(tableName)){
            currentCommitPageSize = commitPageSize;
            if(isDelete) deleteTableData(tableName);
            //通用表处理
            long tableRowNumer = countTable(originConnection,tableName);
            System.out.print("\r");
            System.out.println(tableName+"--->开始同步" +",共计："+tableRowNumer+" 条数据"+(tableRowNumer>1000?"（数据较多时请耐心等待）":""));
            startTime = System.currentTimeMillis();
            currentDealTableName = tableName;
            currentDealTableRowNumer = tableRowNumer;
            if(tableRowNumer==0) {
                System.out.print("\r");
                System.out.println(tableName+"--->同步完成");
                haveSyncTableNameList.add(tableName);
                return;
            }
            String selectSQL = "select * from " + tableName;
            syncWithPage(tableName, selectSQL);
        }
        else{

            //特殊表处理
            String config = specialTableMap.get(tableName);
            JSONObject configJSON = JSON.parseObject(config);
            String targetTable = configJSON.getString("targetTable");
            String selectSQL = configJSON.getString("selectSQL");
            String thisCommitPageSize = configJSON.getString("commitPageSize");
            currentCommitPageSize = (thisCommitPageSize==null || "".equals(thisCommitPageSize))?commitPageSize:Integer.parseInt(thisCommitPageSize);

            if(isDelete) deleteTableData(targetTable);
            long tableRowNumer = countTable(originConnection,tableName);
            System.out.print("\r");
            System.out.println(tableName+"--->开始同步" +",共计："+tableRowNumer+" 条数据"+(tableRowNumer>1000?"（数据较多或涉及Clob字段较多时，耗时较长，请耐心等待！）":""));
            startTime = System.currentTimeMillis();
            currentDealTableName = targetTable;
            currentDealTableRowNumer = tableRowNumer;
            syncWithPage(targetTable, selectSQL);
        }
        haveSyncTableNameList.add(tableName);
        endTime = System.currentTimeMillis();
        System.out.print("\r");
        System.out.println(tableName+"--->同步完成,耗时："+(endTime-startTime)+"ms");
    }

    //删除表数据
    public static void deleteTableData(String tableName) throws SQLException {
        String tableSizeSQL = "delete from " + tableName;
        PreparedStatement preparedStatement = targetConnection.prepareStatement(tableSizeSQL);
        preparedStatement.execute();
    }

    //查询表数据条数
    public static long countTable(Connection connection,String tableName) throws SQLException {
        String tableSizeSQL = "select count(1) as num from " + tableName;
        PreparedStatement preparedStatement = connection.prepareStatement(tableSizeSQL);
        ResultSet resultSet = preparedStatement.executeQuery();
        ResultSetMetaData metaData = resultSet.getMetaData();
        String columnName = metaData.getColumnName(1);
        resultSet.next();
        return resultSet.getLong(columnName);
    }

    //分页同步
    public static void syncWithPage(String toTableName,String selectSQL) throws Exception {
        StringBuilder baseSql = new StringBuilder("insert into " + toTableName + "(");
        statement = originConnection.prepareStatement(selectSQL);
        resultSetTableData = null;
        resultSetTableData = statement.executeQuery();
        int counter = 0;//计数器
        PreparedStatement preparedStatement = null;
        boolean baseSqlIsFull = false;
        while (resultSetTableData.next()) {
            ResultSetMetaData metaData = resultSetTableData.getMetaData();
            int columnCount = metaData.getColumnCount();
            List<Object> valueList = new ArrayList<>();
            for (int i = 1; i <= columnCount; i++) {
                String columnName = metaData.getColumnName(i);
                if(!baseSqlIsFull){
                    if (i == columnCount) {
                        baseSql.append(columnName).append(") values(");
                        for (int j = 1; j <= columnCount; j++) {
                            if (j == columnCount) {
                                baseSql.append("?)");
                            } else {
                                baseSql.append("?,");
                            }
                        }
                    } else {
                        baseSql.append(columnName).append(",");
                    }
                    preparedStatement = targetConnection.prepareStatement(baseSql.toString());
                }
                Object obj = resultSetTableData.getObject(columnName);
                valueList.add(obj);
            }
            baseSqlIsFull = true;
            doInsert(preparedStatement,valueList);
            counter++;
            //批量添加
            if((counter % currentCommitPageSize)==0){
                preparedStatement.executeBatch();
                preparedStatement.clearBatch();
                targetConnection.commit();
                counter = 0;
            }
        }

        preparedStatement.executeBatch();
        preparedStatement.clearBatch();
        targetConnection.commit();
        resultSetTableData.close();
        statement.close();
    }

    //添加数据
    public static void doInsert(PreparedStatement preparedStatement,List<Object> param) throws SQLException {
        for(int i=0;i<param.size();i++) {
            Object o = param.get(i);
            if(o instanceof String){
                String str = ((String) o).trim();
                preparedStatement.setObject(i + 1,str);
            }
            else if(o instanceof Clob){
                String str =  getClobString(o);
                preparedStatement.setObject(i + 1,str);
            }
            else {
                preparedStatement.setObject(i + 1, param.get(i));
            }
        }
        preparedStatement.addBatch();
    }

    //处理clob
    public static String getClobString(Object obj) throws SQLException {
        if(obj instanceof Clob){
            Clob clob = (Clob) obj;
            /*
             * clob长度为4193时，会报违反协议，未知错误，暂分段处理
             */
            if((int)clob.length()==4193){
                return clob.getSubString(1, 4000 - 1) +
                        clob.getSubString(4000, (int) clob.length() - 4000) +
                        clob.getSubString((int) clob.length(), 1);
            }
            return clob.getSubString(1, (int)clob.length());
        }else{
            return obj == null?"":(String)obj;
        }
    }

    //进度查询定时任务
    static class ChechScheduleTimerTask extends TimerTask{
        @Override
        public void run() {
            //通用表处理
            try {
                if(currentDealTableName==null) return;
                long haveSyncRowNumer = countTable(targetConnection,currentDealTableName);
                if(haveSyncRowNumer>0){
                    System.out.print("\r");
                    System.out.print(currentDealTableName+"--->同步进度："+haveSyncRowNumer+"/"+currentDealTableRowNumer);
                    System.out.flush();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }
}
